package com.py.custom.ability.log.plugin;

import com.py.custom.ability.log.kafka.CustomKafkaProducer;
import org.apache.logging.log4j.core.Filter;
import org.apache.logging.log4j.core.Layout;
import org.apache.logging.log4j.core.LogEvent;
import org.apache.logging.log4j.core.appender.AbstractAppender;
import org.apache.logging.log4j.core.config.plugins.Plugin;
import org.apache.logging.log4j.core.config.plugins.PluginAttribute;
import org.apache.logging.log4j.core.config.plugins.PluginElement;
import org.apache.logging.log4j.core.config.plugins.PluginFactory;
import org.apache.logging.log4j.core.layout.PatternLayout;

import java.io.Serializable;

/**
 * 日志收集处理类
 */
@Plugin(name = "CustomKafkaAppender", category = "Core", elementType = "appender", printObject = true)
public class CustomKafkaAppender extends AbstractAppender {

    /**
     * Kafka 生产者
     */
    private CustomKafkaProducer kafkaProducer;

    /**
     * 消息格式化对象
     */
    private Layout<? extends Serializable> layout;

    protected CustomKafkaAppender(String topicName, String bootstrapServers, String name,
                                  Filter filter, Layout<? extends Serializable> layout) {
        super(name, filter, layout);
        this.layout = layout;
        // 初始化kafka发送者
        this.kafkaProducer = new CustomKafkaProducer(topicName, bootstrapServers);
    }

    /**
     * 核心收集方法
     *
     * @param event
     */
    @Override
    public void append(LogEvent event) {
        byte[] bytes = layout.toByteArray(event);
        String logStr = new String(bytes);
        kafkaProducer.sendLog(logStr);
    }

    /**
     * log4j2 使用 appender 插件工厂，因此传参可以直接通过 PluginAttribute 注解注入
     *
     * @param topicName
     * @param bootstrapServers
     * @param name
     * @param filter
     * @param layout
     * @return
     */
    @PluginFactory
    public static CustomKafkaAppender createAppender(@PluginAttribute("topicName") String topicName,
                                                     @PluginAttribute("bootstrapServers") String bootstrapServers,
                                                     @PluginAttribute("name") String name,
                                                     @PluginElement("Filter") final Filter filter,
                                                     @PluginElement("Layout") Layout<? extends Serializable> layout) {
        String format = "%d{yyyy-MM-dd HH:mm:ss} %X{traceId} [%thread] %-5level %logger{36}: %msg%n";
        if (layout != null) {
            format = "%X[traceId] " + layout.getContentFormat().get("format");
        }
        PatternLayout patternLayout = PatternLayout.newBuilder()
                .withPattern(format)
                .build();
        return new CustomKafkaAppender(topicName, bootstrapServers, name, filter, patternLayout);
    }
}
